package com.wudl.core;

import com.wudl.bean.OrderEvent;
import com.wudl.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @version v1.0
 * @ProjectName Flinklearning
 * @ClassName AccountCheckMonitoring
 * @Description TODO 实时对账---监控
 * @Author wudl
 * @Date 2020/10/27 17:13
 */

public class AccountCheckMonitoring {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        //1.读取数据转化为 bean 对象
        SingleOutputStreamOperator<OrderEvent> orderDs = env.readTextFile("F:\\ideaWorkSpace2020\\jg\\Flinklearning\\wudl-flink-java\\input\\OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {

                         @Override
                         public OrderEvent map(String s) throws Exception {
                             String[] split = s.split(",");
                             return new OrderEvent(Long.valueOf(split[0]), split[1], split[2], Long.valueOf(split[3]));
                         }
                     }

                );

        SingleOutputStreamOperator<TxEvent> txDs = env.readTextFile("F:\\ideaWorkSpace2020\\jg\\Flinklearning\\wudl-flink-java\\input\\ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String s) throws Exception {
                        String[] datas = s.split(",");
                        return new TxEvent(datas[0], datas[1], Long.valueOf(datas[2]));
                    }
                });

        //处理数据进行对账

        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDs.keyBy(order -> order.getTxId()).connect(txDs.keyBy(tx -> tx.getTxId()));

        // 使用process进行处理
        SingleOutputStreamOperator<String> resultDS =  orderTxCS.process(new CoProcessFunction<OrderEvent, TxEvent, String>() {

            // 用来存放 交易系统 的数据
            private Map<String, TxEvent> txMap = new HashMap<>();
            // 用来存放 业务系统 的数据
            private Map<String, OrderEvent> orderMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context context, Collector<String> out) throws Exception {
                TxEvent txEvent = txMap.get(value.getTxId());
                if (txEvent == null) {
                    orderMap.put(value.getTxId(), value);
                } else {
                    // 2.说明 交易数据 来了 => 对账成功
                    out.collect("订单" + value.getOrderId() + "对账成功");
                    txMap.remove(value.getTxId());
                }
            }


            @Override
            public void processElement2(TxEvent value, Context context, Collector<String> out) throws Exception {

                OrderEvent orderEvent = orderMap.get(value.getTxId());
                if (orderEvent == null) {
                    txMap.put(value.getTxId(), value);
                } else {
                    // 2.说明 业务数据 来了 => 对账成功
                    out.collect("订单" + orderEvent.getOrderId() + "对账成功");
                    // 对账成功，将保存的 业务数据 删掉
                    orderMap.remove(value.getTxId());
                }


            }
        });
        resultDS.print();
        env.execute();
    }


}
